package com.data.capture.sink;

import com.data.capture.config.BeanConstant;
import com.data.capture.config.properties.CdcProperties;
import com.data.capture.mq.dto.ChangeDataCaptureInfo;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.beans.BeanUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * 数据变更信息流
 * @author linch
 **/
@Log4j2
public class RocketMqSinkFunction extends RichSinkFunction<ChangeDataCaptureInfo> {

    /**
     * 监听配置（Map&lt;监控数据表, MQ主题&gt;）
     */
    private final Map<String, List<CdcProperties>> monitorMap;

    private RocketMqSinkFunction(Map<String, List<CdcProperties>> monitorMap) {
        this.monitorMap = monitorMap;
    }

    @Override
    public void invoke(ChangeDataCaptureInfo info, Context context) {
        // 将key全部转换为大写，兼容 openGauss 和mysql
        List<CdcProperties> list = monitorMap.get(String.format("%s.%s", info.getDatabase(), info.getTableName()));
        if (Objects.isNull(list)) {
            return;
        }
        info.setTableName(info.getTableName().toUpperCase());
        list.forEach(properties -> {
            if (!properties.needPush(info)) {
                log.info("【数据变更信息流】忽略发送MQ消息 topicName = {} changeFields = {} include={} exclude ={}",
                    properties.getTopicName(), info.getChangeFields(), properties.getIncludeFieldList(), properties.getExcludeFieldList());
                return;
            }
            // 只推送id，剩下的全部删除
            if (properties.getOnlyPushPrimary()) {
                info.toSimple();
            }
            // 发送mq消息
            BeanConstant.mqTemplate().convertAndSend(properties.getTopicName(), info);
            if (log.isDebugEnabled()) {
                log.debug("【数据变更信息流】发送MQ消息 topicName = {} content = {}", properties.getTopicName(), info.toString());
            } else {
                log.info("【数据变更信息流】发送MQ消息 topicName = {} table = {} changeField={}", properties.getTopicName(), info.getTableName(), info.getChangeFields());
            }
        });
    }

    /**
     * 数据变更信息流构建器
     * @return 构建器
     */
    public static RocketMqSinkFunctionBuilder builder() {
        return new RocketMqSinkFunctionBuilder();
    }

    /**
     * 数据变更信息流构建器
     */
    @Setter
    @Accessors(chain = true)
    public static class RocketMqSinkFunctionBuilder {

        /**
         * MQ流水槽配置信息
         */
        private Map<String, CdcProperties> mqSinkFunction;

        /**
         * 构建数据变更信息流
         * @return 数据变更信息流
         */
        public RocketMqSinkFunction build() {
            Map<String, List<CdcProperties>> monitorMap = new HashMap<>();
            mqSinkFunction.forEach((sinkName, properties) -> {
                Arrays.stream(properties.getTableList().split(",")).forEach(tableName -> {
                    CdcProperties tableProperties = new CdcProperties();
                    BeanUtils.copyProperties(properties, tableProperties);
                    monitorMap.putIfAbsent(tableName, new ArrayList<>());
                    List<CdcProperties> list = monitorMap.get(tableName);
                    String confTableName = tableName.split("\\.")[1];
                    tableProperties.setIncludeFieldList(loadFieldList(confTableName, properties.getIncludeField(), "include"));
                    tableProperties.setExcludeFieldList(loadFieldList(confTableName, properties.getExcludeField(), "exclude"));
                    tableProperties.setIncludeField(null);
                    tableProperties.setExcludeField(null);
                    list.add(tableProperties);
                });
            });
            return new RocketMqSinkFunction(monitorMap);
        }


        /**
         * 加载字段配置
         * 格式：T_DZ=ZHHSSJ,ZPID;T_RK_RFGL=ZHHSSJ;
         */
        Set<String> loadFieldList(String tableName, String fieldStr, String type) {
            if (StringUtils.isBlank(fieldStr)) {
                return new HashSet<>();
            }
            try {
                Set<String> fieldList = new HashSet<>();
                for (String tableConfig : fieldStr.split(";")) {
                    String configTable = tableConfig.split("=")[0];
                    String configField = tableConfig.split("=")[1];
                    if (!tableName.equals(configTable)) {
                        continue;
                    }
                    // 加入字段
                    fieldList.addAll(Arrays.asList(configField.split(",")));
                }
                if (CollectionUtils.isNotEmpty(fieldList)) {
                    log.info("读取到监控配置 tableName = [{}] fieldList = [{}] type = [{}]", tableName, fieldList, type);
                }
                return fieldList;
            } catch (Exception e) {
                throw new IllegalArgumentException("字段配置格式错误,数据：" + fieldStr + "格式：T_DZ=ZHHSSJ,ZPID;T_RK_RFGL=ZHHSSJ");
            }
        }

    }

}
